package com.lzh.eventdemo.queue;

import com.lzh.eventdemo.entity.Task;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @Description: 模拟队列
 * @Author: LiuZiHao
 * @Date: 2022/8/26 11:28
 */
@Component
@Slf4j
public class MockQueue {
    /**
     * 接收队列
     */
    private BlockingQueue<Task<String>> receiveQueue = new LinkedBlockingQueue<>();
    /**
     * 完成队列
     */
    private BlockingQueue<Task<String>> completeQueue = new LinkedBlockingQueue<>();

    /**
     * 接收任务
     */
    public void put(Task<String> task) throws InterruptedException {
        receiveQueue.put(task);
    }

    /**
     * 获取完成任务
     */
    public Task<String> get() throws InterruptedException {
        return completeQueue.take();
    }

    /**
     * 处理任务
     */
    private void execute() {
        new Thread(() -> {
            while (true) {
                try {
                    /**
                     * 从接收队列中取出任务，处理，然后放入成功队列
                     * take()方法：取走BlockingQueue里排在首位的对象,若BlockingQueue为空,[阻断]进入等待状态直到Blocking有新的对象被加入为止
                     * 线程阻塞直到有新对象进入
                     */
                    Task<String> task = receiveQueue.take();
                    log.info("队列收到数据，处理中");
                    Thread.sleep(2000L);
                    task.setMessage("success");
                    /**
                     * 如果超时，中断该任务，此处应该枷锁
                     */
                    if (task.getIsTimeout()) {
                        log.info("任务超时，处理线程中断该任务");
                        continue;
                    }
                    log.info("队列处理完成");
                    completeQueue.put(task);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    /**
     * 开启处理任务
     */
    public MockQueue() {
        execute();
    }
}

